package chinaums.service.impl;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;

import org.apache.commons.io.IOUtils;
import org.apache.http.HttpEntity;
import org.apache.http.client.ClientProtocolException;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;

import redis.clients.jedis.JedisPool;
import chinaums.common.listener.AbstractCfpsMQMessageListener;
import chinaums.common.util.JsonUtils;
import chinaums.entity.Bacrd;
import chinaums.entity.Numbers;
import chinaums.plugin.https.ErrorConstants;
import chinaums.plugin.https.MessageProto;
import chinaums.plugin.https.MsgPackAndUnpack;
import chinaums.plugin.https.OutputMessageProto;
import chinaums.plugin.redis.CacheClient;
import chinaums.service.IBacrdService;
import chinaums.service.IConfigureService;
import chinaums.service.IRuleService;
import chinaums.service.ISptomlinkService;
import chinaums.service.IUserService;
import chinaums.service.NumbersService;

/**
 * @author liyongjie
 * @date 2017年6月14日 下午2:37:47
 * @version v1.0
 * @description 实现AbstractCfpsMQMessageListener编写关于接收mq流水进行处理的各种逻辑操作
 */
@Service
@Transactional
public class MessageListenerImpl extends AbstractCfpsMQMessageListener {
	private Logger logger = LoggerFactory.getLogger(MessageListenerImpl.class);
	@Resource(name = "jedisClientImpl")
	private CacheClient cacheClient;
	@Autowired
	private JedisPool jedisPool;
	@Autowired
	private IUserService userService;
	@Autowired
	private IRuleService ruleService;
	@Autowired
	private IBacrdService bacrdService;
	@Autowired
	private ISptomlinkService sptomlinkService;
	@Autowired
	private IConfigureService IconfigureService;
	@Autowired
	private NumbersService numberService;
	private Map<String, Object> maps;
	private ExecutorService fixedThreadPool;
	private CountDownLatch countDownLatch;
	private long tnum = 0L;
	// 渠道的私钥,这里是编号为10008的渠道的私钥
	private static final String POST_URL = "http://172.30.252.10:9008/api/service";
	private static final String CHANNEL_PRIVATE_KEY = "30820276020100300d06092a864886f70d0101010500048202603082025c0201000281810091c997463138fcb1f3611fc38d39878e665865ffc1fbbfe0bb03eb315a1896f3d98f874d72bb98b49631addcc2e884d9ff09e0299ccfc4ebc7cef52f2059d86fc9bd9a4a92a45969342d84821efeffd58e80045fe3db98e3a8f65aa238fa170890714fdcf00d2d60841c1261c56753c2958f4f6603d4d3e23db5cc42d6a2cbcf020301000102818022b80f2224c3cd0e360759d0b0d7d40c1a1c9041fdf90de39fb619b359a10486133866c10f8354a52125b8a19bfb52f435942d01cbfa010c23bc0a510c53c942ce1142b6aaaff71677b7e6d22ca92b39773addc978393d5ccf28666c6c409c989333db698fd3733f77e534ef56190e4410180a0589d9c20f54cb8abc8ba76cc5024100c4b1d8941555169774b9ba2fd4abc504c8c8ecfbb3afe4737c8896644229b5e32a4f3c8eb2d355b8bebcefb0d5c8280d4c906d49dd5a816cc1586f8bfc04cb8b024100bdbe66043f1e2c22cda8adda9826136e644365a00a4d6b229b3d60d486dbe9af3f6ecdc6fd505c94dd6a24f004dd568431f4f3204b43730ea319d9782213194d02410091315ad70cdc2f7e038ae9d64eddb0e474366540f385fe54057469d77e437de13f25186599b988c883fbc1cf3e2faee90a407e7c0ceab729099bf75354e8513b024040efd180308dc5f31f49543e9485ab6ec357a187151f8bc3eb37255576d5fa72e495aa4d28d280ecdd9dff0839d1fa5301b9e01f3442e034a933268f0bb2579d02403dac1f74b150e868265b8a53f47c8db1580459a60e771369bbcb83e507fde69ed189a4f238493911ec47eec8607378ca762edc7e13a7ebda54c75764ad4003f0";
	// 智慧验证平台的公钥
	private static final String SMAU_PUB_KEY = "30819f300d06092a864886f70d010101050003818d0030818902818100a136bd1399f94e46ff42eb6fad480621ca3e03c01e9fc8e1eab6d75dabc63e47ef1c2064fb10abaa491b9452e2d504de88223775e14931c6d8c7aab0f13d6673f567d7d0b0769539a749d920f50d7421f3fd66b5425bc3dee6f27b82de6243894e79b76a478efe79ff44ee813e789150871498186cbf1672a306469d250b703d0203010001";
	private static final String CONTENT_TYPE_TEXT = "application/json";

	/*
	 * 线程初始化
	 */
	@PostConstruct
	public void init() {
		/*
		 * 初始化线程池
		 */
		// 设置定义开了10个线程
		fixedThreadPool = Executors.newFixedThreadPool(10);
		// 将线程进行托管
		countDownLatch = new CountDownLatch(10);
		// listMapsSize = map.size();// 数量
		// count = listMapsSize / RunSize;// 每个线程执行的数量
		// logger.info("*****************************************"+"进行了初始化…………………………………………………………………………………………");
	}

	@Override
	public void execute(String data) {
		// System.out.println(data);
		/*
		 * 接收流水mq的值data进行json转换map集合放到多线程里面
		 */
		// 接收MQ值转换后存放到list集合中
		maps = JsonUtils.json2Map(data);
		// 将map及其相关的内容放入到线程池里面
		fixedThreadPool.execute(new MessageListrnerRunnable(maps, cacheClient,
				ruleService, IconfigureService, bacrdService, numberService,
				sptomlinkService, countDownLatch));
	}

	/*
	 * 编写线程方法实现Runnable方法把需要存放到多线程的代码放在run()方法里面
	 */
	public class MessageListrnerRunnable implements Runnable {
		/*
		 * 1，实例化所有类
		 * 2，在run()方法里面进行逻辑判断：当银行卡号与redis的银行卡号相同的时候白名单匹配成功，获取到手机号码和规则编号存入到表中；
		 * 当消费金额大于500的时候获取手机号码和规则编号存入到数据库表中。
		 */
		@Resource(name = "jedisClientImpl")
		private CacheClient cacheClient;
		@Autowired
		private IRuleService ruleService;
		@Autowired
		private IConfigureService IconfigureService;
		@Autowired
		private IBacrdService bacrdService;
		@Autowired
		private ISptomlinkService sptomlinkService;
		@Autowired
		private NumbersService numberService;
		private CountDownLatch countDownLatch;
		private Map<String, Object> maps;
		private long a;
		private String phoneNo = null;

		// 对相关库类进行构造操作
		public MessageListrnerRunnable(Map<String, Object> maps,
				CacheClient cacheClient, IRuleService ruleService,
				IConfigureService iconfigureService,
				IBacrdService bacrdService, NumbersService numberService,
				ISptomlinkService sptomlinkService,
				CountDownLatch countDownLatch) {
			this.maps = maps;
			this.countDownLatch = countDownLatch;
			this.bacrdService = bacrdService;
			this.IconfigureService = iconfigureService;
			this.cacheClient = cacheClient;
			this.ruleService = ruleService;
			this.numberService = numberService;
			this.sptomlinkService = sptomlinkService;
		}

		public String getRequestMsg() throws Exception {
			Map<String, String> body = new HashMap<String, String>();
			// 获取流水的银行卡号
			String cardNo = String.valueOf(maps.get("pri_acct_no"));
			// 把银行卡号放入到报文中
			body.put("cardNo", cardNo);
			// 将请求报文相应的字段按照顺序添加进去
			String msg = MsgPackAndUnpack.getRequestMsg(body,
					CHANNEL_PRIVATE_KEY, "10008", "BDPG01", "test");
			System.out.println("请求报文是：" + msg);
			//将请求的报文以日志的打出
			logger.info("请求的报文是：" + msg);
			return msg;
		}

		/*
		 * 编写run()方法进行逻辑操作代码(non-Javadoc)
		 * 
		 * @see java.lang.Runnable#run()
		 */
		/**
		 * 解析返回结果报文,获取报文中的业务字段
		 * 
		 * @param respMsg
		 */
		public void parseResult(String respMsg) {
			MessageProto msg = JSONObject.parseObject(respMsg,
					MessageProto.class);
			// 通过MsgHead里面的字段判断请求是否成功
			if (ErrorConstants.ERROR_CODE_NO_ERROR.equals(msg.getMsgHead()
					.getMsgRspCode())) {
				// 调用成功，返回报文中有MsgBody
				JSONObject msgBody = (JSONObject) msg.getMsgBody();
				// 获取业务数据返回结果，从中获取到手机号码以及更新时间
				phoneNo = msgBody.getString("phoneNo");
				String updateTime = msgBody.getString("updateTime");
				System.out.println("手机为：" + phoneNo);
				System.out.println("更新时间为:" + updateTime);
				// logger.info("手机为："+phoneNo);
				// logger.info("更新时间为："+updateTime);

			} else if (ErrorConstants.ERROR_CODE_PARAMETER_ERROR.equals(msg
					.getMsgHead().getMsgRspCode())) {
				// 输入参数校验错误，如卡号格式不正确，手机号格式不正确，缺少必填参数等等，返回报文中无MsgBody
				// 此时不能够通过msg.getMsgBody()获取响应体,应该提示客户端msg.getMsgHead().getMsgRspResult();
				System.out.println(msg.getMsgHead().getMsgRspResult());
			} else if (ErrorConstants.ERROR_CODE_TOO_FREQUENT_ACCESS.equals(msg
					.getMsgHead().getMsgRspCode())) {
				// 接口调用频率超出限制，返回报文中无MsgBody
				System.out.println(msg.getMsgHead().getMsgRspResult());
			} else {
				// 其他错误，返回报文中无MsgBody
				System.out.println(msg.getMsgHead().getMsgRspResult());
			}
		}

		@Override
		public void run() {
			// 获取当前时间的毫秒数
			long as = System.currentTimeMillis();
			// TODO: 2017/7/13 try catch 异常捕捉好 确保数据得到处理
			String postMsg = null;
			try {
				// 调用请求报文
				postMsg = getRequestMsg();
			} catch (Exception e1) {
				// TODO Auto-generated catch block
				e1.printStackTrace();
			}
			// 创建调用接口连接
			CloseableHttpClient httpclient = HttpClients.custom().build();
			try {
				// 获取请求响应地址
				HttpPost httpPost = new HttpPost(POST_URL);
				System.out.println("Executing request "
						+ httpPost.getRequestLine());
				//设置编码格式为utf-8
				StringEntity se = new StringEntity(postMsg, "UTF-8");
				se.setContentType(CONTENT_TYPE_TEXT);
				se.setContentEncoding("UTF-8");
				httpPost.setEntity(se);

				CloseableHttpResponse response = httpclient.execute(httpPost);
				try {
					HttpEntity entity = response.getEntity();
					System.out.println(response.getStatusLine());
					if (200 == (response.getStatusLine().getStatusCode())) {
						// String respStr =
						// IOUtils.toString(entity.getContent());
						// 获取相应的报文并将其转换为utf-8格式
						String respStr = IOUtils.toString(
								 entity.getContent(), "UTF-8");
						System.out.println(respStr);
						// 将得到的报文进行日志打出
						logger.info("获取得到的报文是：" + respStr);
						// 进行报文签名的验证，只接受签名验证通过的报文
						boolean result = MsgPackAndUnpack.verify(respStr,
								SMAU_PUB_KEY);
						System.out.println("结果报文签名验证结果为:" + result);
						// 解析返回结果报文,获取报文中的业务字段
						parseResult(respStr);
					}
				} catch (Exception e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				} finally {
					response.close();
				}
			} catch (ClientProtocolException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} finally {
				try {
					httpclient.close();
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}

			// 定义时间获取当前时间
			try {
				Date d = new Date();
				SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
				// 获取到当前时间
				String dateNowStr = sdf.format(d);
				// 获取银行卡号
				String settle_bacrd = String.valueOf(maps.get("pri_acct_no"));
				// String NewSettle_bacrd =
				// String.valueOf("QRSMS_MQ_"+maps.get("pri_acct_no"));
				// System.out.println("银行卡号为："+settle_bacrd);
				// System.out.println("拼加后的银行卡号为："+NewSettle_bacrd);
				// 获取消费金额
				int trans_at = Integer.valueOf(String.valueOf(maps
						.get("trans_at")));
				// System.out.println("^^^^^^^^^^^^^^^^^^^^^^^^^^"+settle_bacrd);
				// System.out.println("^^^^^^^^^^^^^^^^^^^^^^^^^^"+trans_at);
				// 规则1,判断银行卡号是否在白名单中
				String redisResult = cacheClient
						.get("QRSMS_MQ_" + settle_bacrd);
				if (redisResult != null && !"".equals(redisResult)) {
					// 进行查询规则1的触发规则编号获取手机号码和规则编号
					// 获取规则编号
					String rule = String.valueOf(ruleService
							.select_numbers("1").get("R_NUMBER"));

					// System.out.println("++++++++++++++++++++++++++++++规则编号="+rule);

					// 通过规则编号查找到生命周期
					String date = IconfigureService.getnumber(rule);
					// System.out.println("**************************************"+
					// date);
					String a[] = date.split("-");
					try {
						// 获取生命周期的第一个时间
						Date date1 = sdf.parse(a[0]);
						// 获取生命周期的第二个时间
						Date date2 = sdf.parse(a[1]);
						// 获取当前时间
						Date d1 = sdf.parse(dateNowStr);
						// 判断本地时间与规则编号的生命周期进行对比 在生命周期内进行流水获取是手机号与规则编号
						if (d1.getTime() > date1.getTime()
								&& d1.getTime() < date2.getTime()) {
							if (rule != null && !"".equals(rule)) {
								// 从报文中获取到手机号码并储存至表中
								/*
								 * Bacrd phone = bacrdService
								 * .select_phone(settle_bacrd);
								 */
								if (phoneNo != null && !"".equals(phoneNo)) {
									// 把手机号和规则编号存放到表中
									Numbers nums = new Numbers();
									nums.setR_number(rule.toString());
									// nums.setB_phone(phone.getB_phone().toString());
									nums.setB_phone(phoneNo);
									numberService.insertall(nums);
									System.out.println("白名单匹配成功，规则编号为："
											+ rule.toString() + "手机号码是："
											+ phoneNo.toString());
								}
							}
						} else {
							IconfigureService.update_number(rule);
						}
					} catch (java.text.ParseException e) {
						e.printStackTrace();
					}
				}
				// 规则2
				if (trans_at > 500) {
					// 进行查询规则2的触发规则编号获取手机号码和规则编号
					// 规则编号
					String rule = String.valueOf(ruleService
							.select_numbers("2").get("R_NUMBER"));

					// System.out.println("++++++++++++++++++++++++++++++规则编号="+rule);

					// 通过规则编号查找到生命周期
					String date = IconfigureService.getnumber(rule);
					String a[] = date.split("-");
					if (rule != null && !"".equals(rule)) {
						try {
							Date date1 = sdf.parse(a[0]);
							// 获取生命周期的第二个时间
							Date date2 = sdf.parse(a[1]);
							// 获取当前时间
							Date d1 = sdf.parse(dateNowStr);
							// 判断本地时间与规则编号的生命周期进行对比 在生命周期内进行流水获取是手机号与规则编号
							if (d1.getTime() > date1.getTime()
									&& d1.getTime() < date2.getTime()) {
								// 手机号
								if (phoneNo != null && !"".equals(phoneNo)) {
									// 将手机号码及规则编号存放到数据库表中
									Numbers nums = new Numbers();
									nums.setR_number(rule.toString());
									// nums.setB_phone(phone.getB_phone().toString());
									nums.setB_phone(phoneNo.toString());
									numberService.insertall(nums);
									System.out.println("消费金额大于500，规则编号为："
											+ rule.toString() + "手机号码是："
											+ phoneNo.toString());// phone.getB_phone().toString())
								}
							} else {
								IconfigureService.update_number(rule);
							}

						} catch (java.text.ParseException e) {
							// TODO Auto-generated catch block
							e.printStackTrace();
						}
					}
				}

			} catch (Exception e) {
				e.printStackTrace();
			} finally {
				countDownLatch.countDown();
			}
			// 获取当前时间的毫秒数
			long ad = System.currentTimeMillis();
			// 获取两个毫秒数之间的时间差
			long df = (ad - as);
			System.out.println("单个耗时：**********" + df);
			tnum = tnum + df;
			System.out.println("=========总耗时：" + tnum);

		}

	}

	/*
	 * 当销毁的时候关闭线程
	 */
	@PreDestroy
	public void dostory() {
		// 关闭线程
		fixedThreadPool.shutdown();
	}

	@Override
	public void stopEvent() {
		// TODO Auto-generated method stub
	}

	@Override
	public boolean isOk() {
		// TODO Auto-generated method stub
		return false;
	}
}